package com.cqrcb.cloud.entity.constant;

import com.cqrcb.cloud.entity.api.thread.CustomThread;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.*;

/**
 * @author ：yangxueyong
 * @date ：Created in 2021/6/1 14:50
 */
@Slf4j
public class ThreadConstant {
    private static Map<String, CopyOnWriteArrayList<CustomThread>> concurrentHashMap = new ConcurrentHashMap<>();
    private static BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(10000);
    private final static RejectedExecutionHandler handler = new DiscardOldestPolicy();
    private final static ThreadPoolExecutor quotaAyncPoolExecutor = new ThreadPoolExecutor(5, 20, 2, TimeUnit.SECONDS, queue,handler);
    private final static int N = 2;

    /**
     * 将额度扣减的任务加入到队列中，需移除老旧任务
     * （待优化部分：concurrentHashMap的容量未清理，应当定期清理该集合）
     *
     * @param customThread 自定义线程
     */
    public static <T extends CustomThread> void addQueue(T customThread) {
        if(customThread == null){
            return;
        }
        String dataId = customThread.getDataId();
        String dataType = customThread.getDataType();
        String type = customThread.getType();
        String operationType = customThread.getOperationType();
        String key =  dataType+ ":" + dataId+ ":" + type + ":" + operationType;
        try {
            CopyOnWriteArrayList<CustomThread> customThreads = concurrentHashMap.get(key);
            if (customThreads == null) {
                synchronized (key) {
                    if (customThreads == null) {
                        customThreads = new CopyOnWriteArrayList<>();
                        concurrentHashMap.put(key, customThreads);
                    }
                }
            }
            if(customThreads.size() > N){
                CopyOnWriteArrayList<CustomThread> delCustomThreads = new CopyOnWriteArrayList<>();
                for (int i = 0; i < customThreads.size() - N ; i++) {
                    CustomThread e = customThreads.get(i);
                    e.setType(null);
                    delCustomThreads.add(e);
                }
                customThreads.removeAll(delCustomThreads);
            }
            customThreads.add(customThread);
        }catch (Exception e){
            log.warn("新增list队列失败",e);
        }
        quotaAyncPoolExecutor.execute(customThread);
    }

    public static <T extends CustomThread> void rmQueue(T customThread){
        if(customThread == null){
            return;
        }
        String dataId = customThread.getDataId();
        String dataType = customThread.getDataType();
        String type = customThread.getType();
        String operationType = customThread.getOperationType();
        String key =  dataType+ ":" + dataId+ ":" + type + ":" + operationType;
        CopyOnWriteArrayList<CustomThread> customThreads = concurrentHashMap.get(key);
        //多线程下 如下代码很可能会报错，因此将其用try catch包起来
        try {
            customThreads.remove(customThread);
        }catch (Exception e){
            log.warn("删除list队列失败",e);
        }
    }

    /**
     * 自定义的拒绝策略，目前没用
     *
     * @author yangxueyong
     * @date 2021/06/03
     */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        public DiscardOldestPolicy() { }

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }
    }

    /**
     * 线程池放弃等待队列中最旧的未处理任务
     */
//    public final static RejectedExecutionHandler handler = new DiscardOldestPolicy();
//    public final static ThreadPoolExecutor quotaAyncPoolExecutor = new ThreadPoolExecutor(2, 5, 2, TimeUnit.SECONDS, new LinkedBlockingQueue<>(2),handler);
//    public static void addQuotaAyncUpdate(Thread thread){
//        quotaAyncPoolExecutor.execute(thread);
//    }
//
//    public static class DiscardOldestPolicy implements RejectedExecutionHandler {
//        /**
//         * Creates a {@code DiscardOldestPolicy} for the given executor.
//         */
//        public DiscardOldestPolicy() { }
//
//        /**
//         * Obtains and ignores the next task that the executor
//         * would otherwise execute, if one is immediately available,
//         * and then retries execution of task r, unless the executor
//         * is shut down, in which case task r is instead discarded.
//         *
//         * @param r the runnable task requested to be executed
//         * @param e the executor attempting to execute this task
//         */
//        @Override
//        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
//            BlockingQueue<Runnable> queue = e.getQueue();
//            if (!e.isShutdown()) {
//                e.getQueue().poll();
//                e.execute(r);
//            }
//        }
//    }
}
